package org.aceor.mddal.gateway.resultset;

import com.google.common.base.Objects;
import com.google.common.collect.Lists;
import org.aceor.mddal.common.sqlparser.result.context.MergeContext;
import org.aceor.mddal.common.sqlparser.result.object.AggregationColumn;
import org.aceor.mddal.common.executor.ExecuteUnit;
import org.aceor.mddal.gateway.executor.ExecutorEngineFactory;
import org.aceor.mddal.common.executor.MergeUnit;
import org.aceor.mddal.gateway.merger.common.ResultSetQueryIndex;
import org.aceor.mddal.gateway.merger.function.AggregateFunction;
import org.aceor.mddal.gateway.merger.function.AggregateValue;
import org.aceor.mddal.gateway.util.AggregateFunctionUtil;
import org.apache.commons.collections.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.util.Collection;
import java.util.List;

/**
 * Created by lxue on 16/4/16.
 */
public class AggregateResultSet extends BaseResultSet implements ResultSet {

    private static final Logger log = LoggerFactory.getLogger(AggregateResultSet.class);

    private AggregateValue currentAggregateResultSet;
    private List<AggregationColumn> aggregationColumns;
    private final ResultSetMetaData resultSetMetaData;
    private final List<String> columnLabels;

    public AggregateResultSet(List<ResultSet> resultSets, MergeContext mergeContext) throws SQLException {
        super(resultSets, mergeContext.getLimit());
        aggregationColumns = mergeContext.getAggregationColumns();
        resultSetMetaData = getResultSets().iterator().next().getMetaData();

        /**
         * resultSetMetaData的操作会抛SQLException,就不放到MergeUnit中了.这里计算出来
         */
        columnLabels = Lists.newLinkedList();
        for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
            columnLabels.add(resultSetMetaData.getColumnLabel(i));
        }
    }

    @Override
    protected boolean nextRowSet() throws SQLException {
        //In aggregated query without GROUP BY,select list only can use aggregated column ==> 只有聚集列,一列一列算出来(合并)就行
        if (null == currentAggregateResultSet) {
            currentAggregateResultSet = calAggregateResultSet();
            return true;
        }
        return false;
    }

    private AggregateValue calAggregateResultSet() {
        ExecuteUnit<ResultSet, AggregateValue> executeUnit = new ExecuteUnit<ResultSet, AggregateValue>() {
            @Override
            public AggregateValue execute(ResultSet input) throws Exception {
                AggregateValue aggregateValue = new AggregateValue();
                //只有一条记录
                if (input.next()) {
                    for (int i = 1; i <= columnLabels.size(); i++) {
                        aggregateValue.put(i, columnLabels.get(i - 1), input.getObject(i));
                    }
                }
                return aggregateValue;
            }
        };
        MergeUnit<AggregateValue, AggregateValue> mergeUnit = new MergeUnit<AggregateValue, AggregateValue>() {
            @Override
            public AggregateValue merge(List<AggregateValue> params) {
                AggregateValue aggregateValue = new AggregateValue();
                for (int i = 1; i <= columnLabels.size(); i++) {
                    String columnLabel = columnLabels.get(i - 1);
                    AggregationColumn aggregationColumn = findAggregationColumn(columnLabel);
                    if (null != aggregationColumn) {
                        Object value = aggregate(aggregationColumn, new ResultSetQueryIndex(columnLabel), params);
                        aggregateValue.put(i, columnLabel, value);
                    } else {
                        log.warn("未找到聚集列 :" + columnLabel);
                    }
                }
                return aggregateValue;
            }
        };
        return ExecutorEngineFactory.getInstance().execute(getResultSets(), executeUnit, mergeUnit);
    }

    public AggregateValue getCurrentAggregateResultSet() {
        return currentAggregateResultSet;
    }

    private Object aggregate(AggregationColumn aggregationColumn, ResultSetQueryIndex queryIndex, Collection<AggregateValue> aggregateValues) {
        AggregateFunction function = AggregateFunctionUtil
                .createAggregateFunction(
                        aggregationColumn.getAggregationType(),
                        aggregateValues.iterator().next().getValue(queryIndex));
        for (AggregateValue aggregateValue : aggregateValues) {
            Object value = aggregateValue.getValue(queryIndex);
            function.merge(value);
        }
        return function.getResult();
    }

    private AggregationColumn findAggregationColumn(String columnName) {
        if (CollectionUtils.isEmpty(aggregationColumns)) {
            return null;
        }
        for (AggregationColumn aggregationColumn : aggregationColumns) {
            if (Objects.equal(columnName, aggregationColumn.getExpression())) {
                return aggregationColumn;
            }
        }
        return null;
    }
}
